// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: gekim
#include "coin2/feed/navi/reader.h"

bool NaviPerSymbolReader::Read(NaviFeed* out) {
  if (!reader_->Read(out)) {
    return false;
  }

  if (out->has_duplicated_header()) {
    symbol_ = out->duplicated_header().symbol();
    header_.CopyFrom(out->duplicated_header());
  } else if (header_.has_symbol()) {
    out->mutable_duplicated_header()->CopyFrom(header_);
  }
  return true;
}

NaviPerSymbolFeedReader::NaviPerSymbolFeedReader(
    const std::experimental::filesystem::path& per_symbol_root_dir,
    const std::string& machine, const std::string exchange_str,
    const std::set<std::string>& symbols, int64_t warmup,
    int64_t begin_timestamp, /* inclusive */
    int64_t end_timestamp,   /* exclusive */
    size_t max_record_size) {
  std::vector<std::experimental::filesystem::path> paths =
      GetSortedPathList(per_symbol_root_dir, machine, exchange_str, symbols,
                        begin_timestamp - warmup, end_timestamp);
  std::vector<std::unique_ptr<orio::io::Reader<CoinFeed>>> readers;
  for (auto& path : paths) {
    readers.push_back(std::make_unique<IntervalFastFeedReader>(
        std::make_unique<
            NaviToFastfeedConverter>(std::make_unique<NaviPerSymbolReader>(
            std::make_unique<NaviFeedProtoReader>(
                std::unique_ptr<orio::io::Reader<orio::record::SimpleRecord>>(
                    orio::record::SimpleReader::FromFile(
                        path, max_record_size,
                        orio::record::SimpleRecordType::LittleEndian)),
                max_record_size))),
        begin_timestamp - warmup, end_timestamp));
    ;
  }

  reader_ = std::make_unique<MergedFastFeedReader>(std::move(readers));
}

std::vector<std::experimental::filesystem::path>
NaviPerSymbolFeedReader::GetSortedPathList(
    const std::experimental::filesystem::path& per_symbol_root_dir,
    const std::string& machine, const std::string& exchange_str,
    const std::set<std::string>& symbols, int64_t begin_timestamp,
    int64_t end_timestamp /* exclusive */) {
  std::vector<std::pair<std::string, std::experimental::filesystem::path>>
      paths;

  auto p = PtimeFromTimestamp(begin_timestamp);
  begin_timestamp = TimestampFromPtime(PtimeFromDatetime(
      p.date().year(), p.date().month(), p.date().day(), 0, 0, 0, 0));

  for (int64_t ts = begin_timestamp; ts < end_timestamp;
       ts += 24 * 60 * 60 * 1'000'000'000LL) {
    auto yyyymmdd = YmdFromTimestamp(ts);
    for (auto& symbol : symbols) {
      auto dir =
          per_symbol_root_dir / machine / exchange_str / yyyymmdd / symbol;
      if (!std::experimental::filesystem::exists(dir)) {
        LOG(WARNING) << "Missing date from stock per symbol feed: " << dir;
        continue;
      }

      for (auto& p : std::experimental::filesystem::directory_iterator(dir)) {
        /// "--" => recipe
        // no "--" => empty recipe
        // 1.morning.20200715-093000+0800--tag..

        if (std::experimental::filesystem::is_directory(p)) {
          continue;
        }
        auto filename = p.path().filename().string();
        auto idx = filename.find("--");
        std::string_view session;
        std::string tag;
        if (idx == std::string::npos) {
          session = filename;
          tag = "";
        } else {
          session = filename.substr(0, idx);
          tag = filename.substr(idx + 2);
        }

        CHECK_THROW(session.size() >= 22);  // 22 = dates + two dots
        std::string datetime(session.substr(session.size() - 20, 20));
        paths.push_back(std::make_pair(datetime, p));
      }
    }
  }

  std::sort(paths.begin(), paths.end());
  std::vector<std::experimental::filesystem::path> out;

  for (const auto& pair : paths) {
    out.push_back(pair.second);
  }

  return out;
}

bool NaviPerSymbolFeedReader::Read(CoinFeed* out) { return reader_->Read(out); }
